package com.example.mq.mqclient;

import com.example.mq.common.BinaryTool;
import com.example.mq.common.Consumer;
import com.example.mq.common.MqException;
import com.example.mq.common.net.*;
import com.example.mq.mqserver.core.BasicProperties;
import com.example.mq.mqserver.core.ExchangeType;
import lombok.extern.slf4j.Slf4j;

import java.io.IOException;
import java.util.Map;
import java.util.UUID;
import java.util.concurrent.ConcurrentHashMap;

@Slf4j
public class Channel {

    private String channelId;

    //记录当前channel所属的连接
    private Connection connection;

    //用来存储 服务器、客户端之间的响应信息<rid,BasicReturns>
    private ConcurrentHashMap<String, BasicReturns>  basicReturnMap = new ConcurrentHashMap<String, BasicReturns>();

    //如果当前channel订阅了队列，需要记录回调
    //此处约定一个channel只能有一个回调
    private Consumer consumer;

    public Channel(String channelId, Connection connection) {
        this.channelId = channelId;
        this.connection = connection;
    }


    public String getChannelId() {
        return channelId;
    }

    public void setChannelId(String channelId) {
        this.channelId = channelId;
    }

    public Connection getConnection() {
        return connection;
    }

    public void setConnection(Connection connection) {
        this.connection = connection;
    }


    public ConcurrentHashMap<String, BasicReturns> getChannels() {
        return basicReturnMap;
    }

    public void setChannels(ConcurrentHashMap<String, BasicReturns> channels) {
        this.basicReturnMap = channels;
    }

    public Consumer getConsumer() {
        return consumer;
    }

    public void setConsumer(Consumer consumer) {
        this.consumer = consumer;
    }

    //当putRetruns rid,BasicReturns消息后，会唤醒当前对象（channel）所有线程，尝试去获取basicReturnMap中的元素，获取成功，
    // 返回这个basicReturns（最后对应的put了的元素也要remvoe掉）
    private BasicReturns waitResult(String rid) throws InterruptedException {
        BasicReturns basicReturns = null;
        while ((basicReturns = basicReturnMap.get(rid)) == null) {
            //包裹还没来如果null，进行阻塞等待
            synchronized (this) {

                //当前对象进入线程等待
                wait();//notify随机唤醒当前对象的一个线程 notifyAll唤醒所有当前对象的线程
            }
        }

        basicReturnMap.remove(rid);//结果收到之后，从集合中删除

        return basicReturns;
    }

    //把服务器响应信息放到 basicReturnMap中，同时提醒当前对象的线程去获取，把响应返回给客户端
    public void putReturns(BasicReturns basicReturns) {
        basicReturnMap.put(basicReturns.getRid(), basicReturns);
        // 当前也不知道有多少个线程在等待上述的这个响应.
        // 把所有的等待的线程都唤醒.
        synchronized (this) {
            this.notifyAll();
        }
    }

    //告诉服务器创建channel
    public boolean createChannel() throws IOException, InterruptedException {
        Request request = new Request();

        BasicArguments args = new BasicArguments();
        args.setChannelId(channelId);
        args.setRid(generateRid());
        byte[] payLoad = BinaryTool.toBytes(args);

        request.setType(0x1);
        request.setLength(payLoad.length);
        request.setPayload(payLoad);

        connection.writeRequest(request);

        //等待服务器响应
        BasicReturns basicReturns = waitResult(args.getRid());

        return basicReturns.isOk();
    }



    private String generateRid() {
        return "Rid-" + UUID.randomUUID();
    }

    //关闭channel
    public boolean close() throws IOException, InterruptedException {

        //创建数据
        BasicArguments args = new BasicArguments();
        args.setRid(generateRid());
        args.setChannelId(channelId);
        byte[] payLoad = BinaryTool.toBytes(args);

        //写入数据
        Request request = new Request();
        request.setType(0x2);
        request.setLength(payLoad.length);
        request.setPayload(payLoad);
        connection.writeRequest(request);

        //等待响应
        BasicReturns basicReturns = waitResult(args.getRid());

        //返回结果
        return basicReturns.isOk();
    }

    //创建交换机
    public boolean exchangeDeclare(String exchangeName
            , ExchangeType exchangeType
            , boolean durable
            , boolean autoDelete
            , Map<String, Object> arguments) throws IOException, InterruptedException {

        //生成交换机类
        ExchangeDeclareArguments arg = new ExchangeDeclareArguments();

        arg.setRid(generateRid());
        arg.setChannelId(channelId);

        arg.setExchangeName(exchangeName);
        arg.setExchangeType(exchangeType);
        arg.setDurable(durable);
        arg.setAutoDelete(autoDelete);
        arg.setArguments(arguments);


        byte[] payLoad = BinaryTool.toBytes(arg);
        Request request = new Request();
        request.setType(0x3);
        request.setLength(payLoad.length);
        request.setPayload(payLoad);

        //写入请求
        connection.writeRequest(request);

        BasicReturns basicReturns = waitResult(arg.getRid());
        return basicReturns.isOk();
    }

    //删除交换机
    public boolean exchangeDelete(String exchangeName) throws IOException, InterruptedException {
        ExchangeDeleteArguments arguments = new ExchangeDeleteArguments();

        arguments.setRid(generateRid());
        arguments.setChannelId(channelId);
        arguments.setExchangeName(exchangeName);
        byte[] payLoad = BinaryTool.toBytes(arguments);


        Request request = new Request();
        request.setType(0x4);
        request.setLength(payLoad.length);
        request.setPayload(payLoad);
        connection.writeRequest(request);
        BasicReturns returns = waitResult(arguments.getRid());

        return returns.isOk();
    }

    //创建队列
    public boolean queueDeclare(
            String queueName
            , boolean durable
            , boolean exclusive
            , boolean autoDelete
            , Map<String, Object> arguments) throws IOException, InterruptedException {
        QueueDeclareArguments arg = new QueueDeclareArguments();
        arg.setRid(generateRid());
        arg.setChannelId(channelId);

        arg.setQueueName(queueName);
        arg.setDurable(durable);
        arg.setExclusive(exclusive);
        arg.setAutoDelete(autoDelete);
        arg.setArguments(arguments);

        byte[] payLoad = BinaryTool.toBytes(arg);

        Request request = new Request();
        request.setType(0x5);
        request.setLength(payLoad.length);
        request.setPayload(payLoad);

        connection.writeRequest(request);

        BasicReturns returns = waitResult(arg.getRid());
        return returns.isOk();
    }

    //删除队列
    public boolean queueDelete(String queueName) throws IOException, InterruptedException {
        QueueDeleteArguments arguments = new QueueDeleteArguments();

        arguments.setRid(generateRid());
        arguments.setChannelId(channelId);
        arguments.setQueueName(queueName);
        byte[] payLoad = BinaryTool.toBytes(arguments);

        Request request = new Request();
        request.setType(0x6);
        request.setLength(payLoad.length);
        request.setPayload(payLoad);

        connection.writeRequest(request);
        BasicReturns returns = waitResult(arguments.getRid());

        return returns.isOk();
    }

    //创建绑定
    public boolean queueBind(String queueName, String exchangeName, String bindingKey) throws IOException, InterruptedException {
        QueueBindArguments arguments = new QueueBindArguments();

        arguments.setRid(generateRid());
        arguments.setChannelId(channelId);
        arguments.setQueueName(queueName);
        arguments.setExchangeName(exchangeName);
        arguments.setBindingKey(bindingKey);
        byte[] payLoad = BinaryTool.toBytes(arguments);

        Request request = new Request();
        request.setType(0x7);
        request.setLength(payLoad.length);
        request.setPayload(payLoad);

        connection.writeRequest(request);
        BasicReturns returns = waitResult(arguments.getRid());
        return returns.isOk();
    }

    //解除绑定
    public boolean queueUnbind(String queueName, String exchangeName) throws IOException, InterruptedException {
        QueueUnbindArguments arguments = new QueueUnbindArguments();

        arguments.setRid(generateRid());
        arguments.setChannelId(channelId);
        arguments.setQueueName(queueName);
        arguments.setExchangeName(exchangeName);

        byte[] payLoad = BinaryTool.toBytes(arguments);
        Request request = new Request();
        request.setType(0x8);
        request.setLength(payLoad.length);
        request.setPayload(payLoad);
        connection.writeRequest(request);

        BasicReturns returns = waitResult(arguments.getRid());
        return returns.isOk();
    }

    //发布消息
    public boolean basicPublish(String exchangeName
            , String routingKey
            , BasicProperties basicProperties
            , byte[] body) throws IOException, InterruptedException {
        BasicPublishArguments basicPublishArguments = new BasicPublishArguments();

        basicPublishArguments.setRid(generateRid());
        basicPublishArguments.setChannelId(channelId);

        basicPublishArguments.setExchangeName(exchangeName);
        basicPublishArguments.setRoutingKey(routingKey);
        basicPublishArguments.setBody(body);

        byte[] payload = BinaryTool.toBytes(basicPublishArguments);
        Request request = new Request();
        request.setType(0x9);
        request.setLength(payload.length);
        request.setPayload(payload);

        connection.writeRequest(request);
        BasicReturns returns = waitResult(basicPublishArguments.getRid());
        System.out.println("basicPublish returns.channelId: "+returns.getChannelId());
        System.out.println("basicPublish returns.isOk: "+returns.isOk());

        return returns.isOk();
    }

    //订阅消息
    //consumer不发送给服务器，本地保存
    public boolean  basicConsume(String queueName
            , boolean autoDelete
            , Consumer consumer) throws IOException, InterruptedException {

        try {
            if(this.consumer!=null){
                //这里我们自己定义一个channel只能有一个回调
                throw new MqException("该 channel 已经设置过消费消息的回调了, 不能重复设置!");
            }
        } catch (MqException e) {
            System.out.println("这里我们自己定义一个channel只能有一个回调");
            e.printStackTrace();
        }
        this.consumer=consumer;
        BasicConsumeArguments arguments = new BasicConsumeArguments();
        arguments.setRid(generateRid());
        arguments.setChannelId(channelId);

        arguments.setQueueName(queueName);
        arguments.setAutoAck(autoDelete);
        arguments.setConsumerTag(channelId);//这里消费消息的时候，设置consumerTag为channelId



        byte[] payLoad = BinaryTool.toBytes(arguments);
        Request request = new Request();
        request.setType(0xa);
        request.setLength(payLoad.length);
        request.setPayload(payLoad);

        connection.writeRequest(request);

        BasicReturns returns = waitResult(arguments.getRid());

        return returns.isOk();
    }

    //

    //确认消息
    public boolean basicAck(String queueName, String messageId) throws IOException, InterruptedException {
        BasicAckArguments arguments = new BasicAckArguments();
        arguments.setRid(generateRid());
        arguments.setChannelId(channelId);

        arguments.setQueueName(queueName);
        arguments.setMessageId(messageId);
        byte[] payLoad = BinaryTool.toBytes(arguments);
        Request request = new Request();
        request.setType(0xb);
        request.setLength(payLoad.length);
        request.setPayload(payLoad);

        connection.writeRequest(request);
        BasicReturns returns = waitResult(arguments.getRid());

        return returns.isOk();
    }

}
